package com.shujia.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet, Timestamp}

object Demo06MapPartitions {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo06MapPartitions")
    conf.setMaster("local")

    val sc: SparkContext = new SparkContext(conf)

    //    val blackStuList: List[String] = List[String]("1500100001", "1500100011", "1500100021")
    // 基于List或者集合类构建RDD
    //    val blackStuRDD: RDD[String] = sc.parallelize(blackStuList)
    val blackStuRDD: RDD[String] = sc.textFile("spark/data/students.txt").map(line => line.split(",")(0))

    /**
     * 如果直接放在算子外部，相当于是放在Driver端
     * 算子内部的代码相当于会在Executor中被Task执行
     *
     * 连接是不能够被序列化的 所以放在算子外部代码运行时会报错：
     * Task not serializable
     * Caused by: java.io.NotSerializableException: java.lang.Object
     *
     */
    // 通过JDBC的方式连接MySQL获取数据
    //    val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")


    // 基于id名单 从MySQL中student表里获取学生基本信息
    //    blackStuRDD
    //      .map(id => {
    //        // 通过JDBC的方式连接MySQL获取数据
    //        val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")
    //        println("建立了一次连接") // 如果直接放在算子内部，则每一条数据都会建立一次连接，太浪费资源了，而且频繁的创建销毁连接会增大MySQL的负载
    //        // 创建statement
    //        val pSt: PreparedStatement = conn.prepareStatement("select id,name,age,gender,clazz,last_mod from student where id = ?")
    //        // 设置参数
    //        pSt.setString(1, id)
    //        // 执行查询
    //        val rs: ResultSet = pSt.executeQuery()
    //
    //        var resStr: String = ""
    //        // 获取数据
    //        while (rs.next()) {
    //          val id: Int = rs.getInt("id")
    //          val name: String = rs.getString("name")
    //          val age: Int = rs.getInt("age")
    //          val gender: String = rs.getString("gender")
    //          val clazz: String = rs.getString("clazz")
    //          val lastMod: Timestamp = rs.getTimestamp("last_mod")
    //          resStr = s"$id,$name,$age,$gender,$clazz,$lastMod"
    //        }
    //        resStr
    //      })
    //      .foreach(println)

    /**
     * 既然无法避免连接被多次创建，那么就尽可能减少连接建立的次数
     * 可以使用mapPartitions让每个Task在处理同一个分区的所有数据时，共用一个连接
     * 即：有多少个分区，就有多少个Task，就有多少个连接
     * mapPartitions也是转换算子，需要action算子触发执行
     *
     * 适用场景：需要从外部系统建立连接然后获取数据时则可以使用mapPartitions来减少连接建立的次数
     */
    blackStuRDD
      // 对每个分区的数据以此处理
      .mapPartitions(iter=>{

        // 通过JDBC的方式连接MySQL获取数据
        val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student?useSSL=false", "root", "123456")
        println("建立了一次连接") // 如果直接放在算子内部，则每一条数据都会建立一次连接，太浪费资源了，而且频繁的创建销毁连接会增大MySQL的负载
        // 创建statement
        val pSt: PreparedStatement = conn.prepareStatement("select id,name,age,gender,clazz,last_mod from student where id = ?")

        /**
         * Spark会将每个分区的数据封装到一个迭代器中
         * 这里使用的map方法实际上是迭代器类型提供的 并不是Spark中的算子
         * 所以map方法中的逻辑不是在算子内部，所以可以直接使用上面建立的连接
         */
        val rdd: Iterator[String] = iter.map(id => {
          // 对每个分区的每条数据进行处理
          // 设置参数
          pSt.setString(1, id)
          // 执行查询
          val rs: ResultSet = pSt.executeQuery()

          var resStr: String = ""
          // 获取数据
          while (rs.next()) {
            val id: Int = rs.getInt("id")
            val name: String = rs.getString("name")
            val age: Int = rs.getInt("age")
            val gender: String = rs.getString("gender")
            val clazz: String = rs.getString("clazz")
            val lastMod: Timestamp = rs.getTimestamp("last_mod")
            resStr = s"$id,$name,$age,$gender,$clazz,$lastMod"
          }
          resStr
        })
        rdd
      })
      .foreach(println)


  }

}
